package com.xiaoshuidi.cloud.module.bpm.job.delayjob;

import com.xiaoshuidi.cloud.framework.tenant.core.context.TenantContextHolder;
import com.xiaoshuidi.cloud.module.bpm.job.delayjob.strategy.IStrategy;
import com.xiaoshuidi.cloud.module.bpm.job.delayjob.strategy.ProcessingTask;
import com.xiaoshuidi.cloud.module.bpm.job.delayjob.dto.TicketDelayJobDto;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.context.annotation.DependsOn;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class TicketDelayJobService {

    private static final String delayJobName = "ticket_delay_job";

    @Resource
    private RedissonClient redissonClient;

    @Resource
    private ProcessingTask processingTask;

    // 实例化一个单线程执行器
    private ExecutorService executorService = Executors.newSingleThreadExecutor();

    @PostConstruct
    @DependsOn(value = {"redissonClient"})
    public void startJobListener() {
        // 启动线程
        ListenerTask task = new ListenerTask(redissonClient);
        executorService.submit(task);
        // 系统重启后,发起一次任务消费, 避免出现队列中有到期但是没有被放到目标队列的任务, 避免任务未处理
        addDelayJob(TicketDelayJobTypeEnum.INIT, "", 1L, TimeUnit.SECONDS);
    }


    /**
     * 添加延迟任务
     */
    public void addDelayJob(TicketDelayJobTypeEnum ticketDelayJobTypeEnum, Object delayJob, Long delay, TimeUnit timeUnit) {
        RBlockingQueue<Object> blockingQueue = redissonClient.getBlockingQueue(delayJobName);
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
        TicketDelayJobDto ticketDelayJobDto = new TicketDelayJobDto().setTicketDelayJobTypeEnum(ticketDelayJobTypeEnum).setDelayJob(delayJob).setTenantId(TenantContextHolder.getTenantId());
        delayedQueue.offer(ticketDelayJobDto, delay, timeUnit);
    }

    class ListenerTask implements Runnable {

        private RedissonClient redissonClient;

        public ListenerTask(RedissonClient redissonClient) {
            this.redissonClient = redissonClient;
        }

        @Override
        public void run() {
            RBlockingQueue<TicketDelayJobDto> blockingQueue = redissonClient.getBlockingQueue(delayJobName);
            while (true) {
                try {
                    TicketDelayJobDto delayJobDto = blockingQueue.take();
                    TenantContextHolder.setTenantId(delayJobDto.getTenantId());
                    if (!TicketDelayJobTypeEnum.INIT.equals(delayJobDto.getTicketDelayJobTypeEnum())) {
                        ProcessingTask processingTask = new ProcessingTask();
                        IStrategy strategy = processingTask.getStrategy(delayJobDto.getTicketDelayJobTypeEnum());
                        strategy.consumerTask(delayJobDto.getDelayJob());
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } finally {
                    TenantContextHolder.clear();
                }
            }
        }
    }
}
